package com.atguigu.flink.datastramapi.transform;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/2/24
 *
 *  Flink和Spark及MR的最明显的差别在哪里?
 *      Spark和MR,不同的阶段，传递数据是使用shuffle机制。
 *      shuffle机制是下游的Task主动到上游的Task去拉取数据！
 *
 *      Flink是上游主动向下游发送数据，下游只需要被动接收。
 *
 *  -----------------------------
 *      socketTextStream, map , print 称为Operator(算子)。
 *
 *      每个Operator(算子)可以并行运行。
 *          并行运行的程度，称为并Parilisim行度。
 *
 *          在idea上，没有设置并行度，那么每个算子会使用集群的最大并行度(CPU核数)。
 *
 *   ----------------------
 *      改：
 *
 *          ①读取配置文件，可以在配置文件中指定
 *              parallelism.default: N
 *
 *          ②提交Job时，可以明确设置
 *               在webUI，可以明确指定
 *               通过命令行提交:  flink run -p 并行度
 *
 *          ③代码中设置
 *                  进行全局设置 env.setParallelism(10);
 *
 *          ④代码中设置
 *                  针对每一个算子，设置并行度
 *                  算子.setParallelism(15)
 *
 *             优先级: ④ > ③ > ② > ①
 *
 *   ------------------------------
 *      分发规则:
 *              所有的分发规则都是 ChannelSelector 的子类。
 *                  void setup(int numberOfChannels);  每个需要向下游分发数据的算子，都有一个ChannelSelector对象，在初始化时，调用
 *                              setup,设置下游的通道数。
 *
 *                   int selectChannel(T record):  当前算子的每一条record，要发往下游的哪个通道。
 *                   boolean isBroadcast()： 是否需要广播。
 *                                                  如果是广播模式，那么将一条数据发往下游的每一个通道。
 *
 *                   有以下子类: 如果希望使用以下分区器，无需自己new，只需要调用对应的方法即可。
 *                   rescale   RescalePartitioner:  在同一个TaskManager内部负载均衡地把上游的数据发送到下游
 *                   reblance   RebalancePartitioner:  绝对的负载均衡地把上游的数据发送到下游.
 *                                  默认。 避免数据倾斜。
 *
 *                   keyBy   KeyGroupPartitioner：  按照key分发，相同key的数据发到下游的同一个通道
 *                   global   GlobalPartitioner:  全局汇总。只会把数据发往下游的第一个通道
 *
 *                   forward   ForwardPartitioner:  1:1直连。 上游0号通道的数据，只会发给下游的0号通道。
 *                                  有条件限制:
 *                                          上游和下游算子的并行度，必须相等!
 *                                          是算子合并(operator chain)的必须条件！
 *
 *                   shuffle   ShufflePartitioner:  随机分发
 *                      CustomPartitionerWrapper: 自定义单播。
 *                   broadcast   BroadcastPartitioner: 广播
 *
 *
 */
public class Demo1_DistributeData
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.disableOperatorChaining();

        //env.setParallelism(10);

        //这个算子是非并行运行的算子。 Parilisim=1 。 所有 SourceFunction类型的source算子，都是非并行运行的！
        //he parallelism of non parallel operator must be 1.
        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888);

        ds.map(x -> x ).setParallelism(3)
           .map(x -> x ).setParallelism(2)
           .map(x -> x ).setParallelism(1)
          //.forward()
          .print().setParallelism(3);

        env.execute();

    }
}
